package com.lsx143.realtime.app;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 用于执行Flink-SQL的基类app
 */
public abstract class BaseSQLApp {
    /**
     * 业务逻辑
     *
     * @param tEnv flink-stream的执性环境
     */
    protected abstract void run(StreamTableEnvironment tEnv);

    /**
     * 初始化配置信息
     *
     * @param port               端口号
     * @param appName            app名称
     * @param defaultParallelism 并行度
     */
    public void init(int port, String appName, int defaultParallelism) {
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", port);
        //设置flink的执行环境参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(defaultParallelism);

        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop162:8020/realtime/" + appName);

        env.enableCheckpointing(3000, CheckpointingMode.EXACTLY_ONCE);

        env.getCheckpointConfig().setCheckpointTimeout(30000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        env.getCheckpointConfig()
                .enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //获取streamTable的执性环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //执行业务逻辑
        System.out.println("[" + appName + "] 启动");
        run(tEnv);

        try {
            env.execute(appName);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
